use crate::{keys::VddKey, CF_CACHE, CF_FV, CF_LOG, CF_META, CF_TRASH, CF_UNIQUE, ROCKSDB};
use rust_rocksdb::WriteBatchWithTransaction;
use visit_dd_core::{Ares, Astr};

fn db_id_to_cf_name(db_id: u8) -> Option<&'static str> {
    match db_id {
        11 => Some(CF_META),
        12 => Some(CF_FV),
        13 => Some(CF_LOG),
        14 => Some(CF_UNIQUE),
        15 => Some(CF_CACHE),
        16 => Some(CF_TRASH),
        _ => None,
    }
}

/// 加载一个数据库中的所有健
pub fn all_keys(db_id: u8, from_index: u64, max_length: usize) -> Vec<String> {
    let mut key_vec = Vec::<String>::new();
    let mut index = 0u64;
    let cf = db_id_to_cf_name(db_id);
    let iter = match cf {
        None => ROCKSDB.prefix_iterator(""),
        Some(cf) => ROCKSDB.prefix_iterator_cf(&ROCKSDB.cf_handle(cf).unwrap(), ""),
    };
    for item in iter {
        let (k, _) = item.unwrap();
        index += 1;
        if index > from_index {
            let key = String::from_utf8(k.to_vec()).unwrap();
            key_vec.push(key);
        }
        if key_vec.len() >= max_length {
            break;
        }
    }
    key_vec
}

pub fn get_value(db_id: u8, key: &[u8]) -> Option<Vec<u8>> {
    let cf = db_id_to_cf_name(db_id);
    match cf {
        None => ROCKSDB.get(key).unwrap(),
        Some(cf) => {
            if cf == CF_UNIQUE || cf == CF_FV {
                Some("二进制数据流，无法展示".as_bytes().to_vec())
            } else {
                let cfh = ROCKSDB.cf_handle(cf).unwrap();
                ROCKSDB.get_cf(&cfh, key).unwrap()
            }
        }
    }
}

pub fn batch_delete(keys: &[Astr], in_trash: bool) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    keys.iter().for_each(|k| {
        if !in_trash {
            let vk = VddKey::try_from(k.as_ref());
            if let Ok(vk2) = vk {
                match vk2.db_name() {
                    Some(cf) => {
                        batch.delete_cf(cf, k.as_ref());
                    }
                    None => {
                        batch.delete(k.as_ref());
                    }
                }
            }
        } else {
            batch.delete_cf(ROCKSDB.cf_handle(CF_TRASH).unwrap(), k.as_ref());
        }
    });
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}

pub fn put_value(db_id: u8, key: &str, value: &str) -> Ares {
    let cf = db_id_to_cf_name(db_id);
    let r = match cf {
        None => ROCKSDB.put(key, value),
        Some(cf) => {
            let cfh = ROCKSDB.cf_handle(cf).unwrap();
            ROCKSDB.put_cf(&cfh, key, &value)
        }
    };
    r.map_err(|e| e.to_string().into())
}

fn remove_last_content_from_key(key: &str) -> String {
    let splited = key.split("/").collect::<Vec<&str>>();
    let key_head = key.strip_suffix(splited.last().unwrap_or(&""));
    key_head.unwrap_or("").to_string()
}

pub fn delete_by_key_head(key: &str, in_trash: bool) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let vk = VddKey::try_from(key)?;
    let kh = &remove_last_content_from_key(&key);
    let iter = if !in_trash {
        match vk.db_name() {
            None => ROCKSDB.prefix_iterator(kh),
            Some(cf) => ROCKSDB.prefix_iterator_cf(cf, kh),
        }
    } else {
        ROCKSDB.prefix_iterator_cf(ROCKSDB.cf_handle(CF_TRASH).unwrap(), kh)
    };
    for x in iter {
        let (key, _) = x.unwrap();
        let key = String::from_utf8(key.to_vec()).unwrap();
        if key.starts_with(kh) {
            if !in_trash {
                match vk.db_name() {
                    None => batch.delete(key),
                    Some(cf) => batch.delete_cf(cf, key),
                }
            } else {
                batch.delete_cf(ROCKSDB.cf_handle(CF_TRASH).unwrap(), key);
            }
        }
    }
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}
